<?php

use Illuminate\Database\Capsule\Manager as DB;

class Queue
{
    protected static $instance;

    protected $options = [
        'expire'  => 60,
        'default' => 'default',
    ];

    public function __construct($options = [])
    {
        if (!empty($options)) {
            $this->options = array_merge($this->options, $options);
        }
    }

    protected function push($job, $data = '', $queue = null)
    {
        if(QUEUE_ENABLE){
            return self::buildQueue()->pushToDatabase(0, $queue, self::buildQueue()->createPayload($job, $data));
        }else{
            set_time_limit(0);
            $payLoad = json_decode(self::buildQueue()->createPayload($job, $data, $queue));
            $syncJob = new QueueJob(self::buildQueue(), $payLoad);
            try {
                $syncJob->syncFire($payLoad);
            } catch (Exception $e) {
                $syncJob->failed();
                throw $e;
            } catch (Throwable $e) {
                $syncJob->failed();
                throw $e;
            }
        }

    }

    protected function createPayload($job, $data = '', $queue = null)
    {
        if (is_object($job)) {
            $payload = json_encode([
                'job'  => 'app\job\callQueuedHandler@call',
                'data' => [
                    'commandName' => get_class($job),
                    'command'     => serialize(clone $job),
                ],
            ]);
        } else {
            $payload = json_encode($this->createPlainPayload($job, $data));
        }

        if (JSON_ERROR_NONE !== json_last_error()) {
            throw new InvalidArgumentException('Unable to create payload: ' . json_last_error_msg());
        }

        return $payload;
    }

    protected function createPlainPayload($job, $data)
    {
        return ['job' => $job, 'data' => $data, 'attempts' => 0];
    }

    protected function later($delay, $job, $data = '', $queue = null)
    {
        if(QUEUE_ENABLE) {
            return $this->pushToDatabase($delay, $queue, $this->createPayload($job, $data));
        }else{
            return $this->push($job, $data, $queue);
        }
    }

    protected function pop($queue = null)
    {
        $queue = self::buildQueue()->getQueue($queue);

        if (!is_null(self::buildQueue()->options['expire'])) {
            self::buildQueue()->releaseJobsThatHaveBeenReservedTooLong($queue);
        }

        $onejob = DB::transaction(function () use ($queue) {
            $job = jobsModel::where('queue', self::buildQueue()->getQueue($queue))
                ->where('reserved', 0)
                ->where('available_at', '<=', time())
                ->orderBy('id', 'asc')
                ->lockForUpdate()
                ->first();
            if ($job) {
                //标记任务正在执行
                $job->reserved    = 1;
                $job->reserved_at = time();
                $job->save();
                return new QueueJob(self::buildQueue(), $job, $queue);
            }
        });

        return $onejob;
    }

    /**
     * 重新发布任务
     * @param  string    $queue
     * @param  \StdClass $job
     * @param  int       $delay
     * @return mixed
     */
    public function release($queue, $job, $delay)
    {
        return $this->pushToDatabase($delay, $queue, $job->payload, $job->attempts);
    }

    /**
     * Push a raw payload to the database with a given delay.
     *
     * @param  \DateTime|int $delay
     * @param  string|null   $queue
     * @param  string        $payload
     * @param  int           $attempts
     * @return mixed
     */
    protected function pushToDatabase($delay, $queue, $payload, $attempts = 0)
    {
        return jobsModel::insert([
            'queue'        => $this->getQueue($queue),
            'payload'      => $payload,
            'attempts'     => $attempts,
            'reserved'     => 0,
            'reserved_at'  => null,
            'available_at' => time() + $delay,
            'created_at'   => date('Y-m-d H:i:s')
        ]);
    }

    /**
     * 重新发布超时的任务
     *
     * @param  string $queue
     * @return void
     */
    protected function releaseJobsThatHaveBeenReservedTooLong($queue)
    {
        $expired = time() - self::buildQueue()->options['expire'];

        jobsModel::where('queue', self::buildQueue()->getQueue($queue))
            ->where('reserved', 1)
            ->where('reserved_at', '<=', $expired)
            ->increment('attempts', 1, [
                'reserved'    => 0,
                'reserved_at' => null,
            ]);
    }

    /**
     * 删除任务
     * @param  string $id
     * @return void
     */
    public function deleteReserved($id)
    {
        jobsModel::where('id', $id)->delete();
    }

    protected function getQueue($queue)
    {
        return $queue ?: $this->options['default'];
    }

    private static function buildQueue()
    {
        if (!isset(self::$instance)) {

            self::$instance = new self();
        }
        return self::$instance;
    }

    public function __call($method, $arg){
        return self::buildQueue()->$method(...$arg);
    }

    public static function __callStatic($method, $arg){
        return self::buildQueue()->$method(...$arg);
    }

}
